package com.atguigu.mapreduce.demo.outputformat;

import com.atguigu.mapreduce.utils.RCloseUtil;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @version V1.0
 * @author: guangcheng
 * @date: 2021/11/30 18:21
 * @description:
 */
public class LogRecordWriter extends RecordWriter<Text, NullWritable> {
    FSDataOutputStream ggOutStream = null;
    FSDataOutputStream otherOutStream = null;

    static String ggPath = "D:\\testarea\\bigdatatest\\hadoop\\mapreduce\\11_input\\inputoutputformat\\gg";
    static String otherPath = "D:\\testarea\\bigdatatest\\hadoop\\mapreduce\\11_input\\inputoutputformat\\other";


    /**
     * 构造环境
     *
     * @param job
     */
    public LogRecordWriter(TaskAttemptContext job) {

        try {
            FileSystem fileSystem = FileSystem.get(job.getConfiguration());
            ggOutStream = fileSystem.create(new Path(ggPath));
            otherOutStream = fileSystem.create(new Path(otherPath));
        } catch (IOException e) {
            System.out.println(e.getMessage());
        } finally {
        }
    }


    /**
     * 业务操作
     *
     * @param key
     * @param value
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {

        String keyStr = key.toString();

        if (keyStr.contains("atguigu")) {
            ggOutStream.writeBytes(keyStr + "\n");
        } else {
            otherOutStream.writeBytes(keyStr + "\n");
        }
    }


    /**
     * 关流
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        RCloseUtil.close(ggOutStream, otherOutStream);
    }
}

